package com.we.mexico.flink.utils;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;

import javax.annotation.Nullable;

/**
 * @author ：Jason
 * @description：Flink 的各种工具类
 */
public class FlinkDataStreamUtils {
    /**
     * 把 Table 转成 DataStream,这里主要是带有回撤流的数据源,直接过滤掉-D 数据
     *
     * @param tEnv StreamTableEnvironment
     * @param query 查询的 SQL 语句
     * @param targetClass 输入的数据类型
     * @return 返回 DataStream
     * @param <T> 想要返回的数据类型
     */
    public static <T> DataStream<T> getRetractStream(
            StreamTableEnvironment tEnv, String query, Class<T> targetClass) {
        Table table = tEnv.sqlQuery(query);
        return tEnv.toRetractStream(table, targetClass)
                .filter(data -> data.f0)
                .map((MapFunction<Tuple2<Boolean, T>, T>) value -> value.f1);
    }

    /**
     * 不带回撤流的 DataStream
     *
     * @param tEnv
     * @param query
     * @param targetClass
     * @return
     * @param <T>
     */
    public static <T> DataStream<T> getDataStream(
            StreamTableEnvironment tEnv, String query, Class<T> targetClass) {
        Table table = tEnv.sqlQuery(query);
        return tEnv.toDataStream(table, targetClass);
    }

    public static <T> DataStream<String> mapToJSONString(
            DataStream<T> dataStream, @Nullable String streamType) {
        return dataStream.map(
                new MapFunction<T, String>() {
                    @Override
                    public String map(T value) throws Exception {
                        final JSONObject jsonObject =
                                JSONObject.parseObject(JSONObject.toJSONString(value));
                        jsonObject.put("stream_type", streamType);
                        return JSONObject.toJSONString(jsonObject);
                    }
                });
    }

    public static <T> DataStream<JSONObject> stringToJsonObject(DataStream<T> dataStream) {
        return dataStream.map(
                new MapFunction<T, JSONObject>() {
                    @Override
                    public JSONObject map(T value) throws Exception {
                        return JSONObject.parseObject(String.valueOf(value));
                    }
                });
    }

    public static <T> DataStream<T> stringToPoJo(
            DataStream<String> dataStream, Class<T> targetClass) {
        return dataStream.map(
                new MapFunction<String, T>() {
                    @Override
                    public T map(String value) throws Exception {
                        return JSON.parseObject(String.valueOf(value), targetClass);
                    }
                });
    }

    public static <T> DataStream<T> filterFuntion(DataStream<T> dataStream, String key) {
        return dataStream.filter(
                new FilterFunction<T>() {
                    @Override
                    public boolean filter(T value) throws Exception {
                        return JSONObject.parseObject(String.valueOf(value)).containsKey(key);
                    }
                });
    }
}
